package lib

import (
	"fmt"
	"log"
	"strings"

	"github.com/streadway/amqp"
)

const (
	ExchangeUser        = "user_exchange" //交换机
	ExchangeUserDelayed = "user_delayed"  //延迟
	RouterKeyUserReg    = "user_reg"      //注册
	QueueNewUser        = "user_new"      //新
	QueueNewUserUnion   = "user_union"    //联合
)

type MQ struct {
	Conn          *amqp.Connection
	Ch            *amqp.Channel
	NotifyConfirm chan amqp.Confirmation //消息发送回执
	notifyReturn  chan amqp.Return       //消息入列回执
}

func NewMQ() *MQ {

	var err error
	m := new(MQ)

	m.Conn, err = amqp.Dial("amqp://admin:admin@192.168.56.103:5672")
	if err != nil {
		fmt.Println(err)
		return nil
	}

	m.Ch, err = m.Conn.Channel()
	if err != nil {
		log.Fatal(err)
	}

	return m
}

//绑定队列
func (m *MQ) DecQueueAndBind(queues string, key string, exchange string) error {

	fmt.Println("-------------queue:", queues)
	qList := strings.Split(queues, ",")

	for k, queue := range qList {

		fmt.Println("--------n:", k, queue)
		//声明
		q, err := m.Ch.QueueDeclare(queue, false, false, false, false, nil)
		if err != nil {
			fmt.Println("队列声明 err: ", err)
			return err
		}
		err = m.Ch.QueueBind(q.Name, key, exchange, false, nil)
		if err != nil {
			return err
		}
	}

	return nil
}

//NotifyReturn 确认消息入列是否成功
func (m *MQ) NotifyReturn() {
	m.notifyReturn = m.Ch.NotifyReturn(make(chan amqp.Return))
	go m.ListenReturn()
}

func (m *MQ) ListenReturn() {
	defer m.Close()
	ref := <-m.notifyReturn
	if string(ref.Body) != "" {
		//TODO 消息入列失败
		log.Println("消息入列失败: ", string(ref.Body))
		return
	}
	log.Println("消息入列成功: ", string(ref.Body))

}

//SetConfirm 消息发送回执
//对性能 有一定的影响
func (m *MQ) SetConfirm() {

	err := m.Ch.Confirm(false)
	if err != nil {
		log.Println(err)
		return
	}

	m.NotifyConfirm = m.Ch.NotifyPublish(make(chan amqp.Confirmation))
}

//ListenConfirm 发送消息确认是否收到
func (m *MQ) ListenConfirm() {

	ref := <-m.NotifyConfirm

	if ref.Ack {
		log.Println("confirm: 发送消息成功")
	} else {
		log.Println("confirm: 发送消息失败")
	}

}

//SendMessage 发送
func (m *MQ) SendMessage(key, exchange, msg string) error {

	err := m.Ch.Publish(exchange, key, true, false, amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte(msg),
	})

	if err != nil {
		fmt.Println("生产出现问题: ", err)
		return err
	}
	return nil
}

//SendDelayedMessage 发送延迟消息
func (m *MQ) SendDelayedMessage(key, exchange, msg string, delay int) error {

	err := m.Ch.Publish(exchange, key, true, false, amqp.Publishing{
		Headers:     map[string]interface{}{"x-delay": delay},
		ContentType: "text/plain",
		Body:        []byte(msg),
	})

	if err != nil {
		fmt.Println("生产出现问题: ", err)
		return err
	}
	return nil
}

//消费者
func (m *MQ) Consume(name, consumer, c string, callback func(<-chan amqp.Delivery, string)) {

	msgs, err := m.Ch.Consume(name, consumer, false, false, false, false, nil)
	if err != nil {
		log.Fatal("消费者 err:", err)
	}

	callback(msgs, c)

}

func (m *MQ) Close() {

	var err error

	err = m.Ch.Close()
	if err != nil {
		fmt.Println(" m.ch.Close: ", err)
		return
	}

	err = m.Conn.Close()
	if err != nil {
		fmt.Println("m.conn.Close: ", err)
		return
	}

}
